/*
    连接模块：连接类
    1. 连接类：连接类相当于muduo库搭建的客户端，弱化了之前客户端的概念，因为现在用户的通信是基于信道，而信道由连接管理
        所以用户通信直接使用连接来获取信道去完成通信即可
        属性：异步工作者线程句柄、muduo客户端、连接句柄、连接同步控制器、请求分发器、协议处理器句柄、信道管理器句柄
        方法：打开信道、关闭信道
*/
#pragma once
#include "proto/codec.h"
#include "proto/dispatcher.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpClient.h"
#include "muduo/base/CountDownLatch.h"
#include "asyncworker.hpp"
#include "channel.hpp"

namespace mq
{
    class Connection
    {
    public:
        using ConnectionPtr = std::shared_ptr<Connection>;
        
    public:
        Connection(const AsyncWorker::AsyncWorkerPtr &async_worker_ptr, int port, const std::string &ip = "127.0.0.1")
            : _async_worker_ptr(async_worker_ptr), _downlatch(1),
              _client(async_worker_ptr->_loopthread.startLoop(), muduo::net::InetAddress(ip, port), "client"),
              _dispatcher(std::bind(&Connection::onUnknownMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
              _codec_ptr(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
              _channel_manager_ptr(std::make_shared<ChannelManager>())
        {
            // 1. 给请求分发器注册业务处理函数
            _dispatcher.registerMessageCallback<CommonResponse>(std::bind(&Connection::onCommonResponse, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<ConsumeResponse>(std::bind(&Connection::onConsumeResponse, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            // 2. 给服务器设置连接建立和断开时的回调 & 消息回调
            _client.setConnectionCallback(std::bind(&Connection::onConnection, this, std::placeholders::_1));
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec_ptr.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            
            // 3. 发起连接
            _client.connect();

            // 4. 等待连接建立完成
            _downlatch.wait();
        }
    public:
        // 打开信道
        Channel::ChannelPtr openChannel()
        {
            Channel::ChannelPtr channel_ptr = _channel_manager_ptr->createChannel(_conn, _codec_ptr); // 当前连接中创建信道
            bool ret = channel_ptr->openChannel(); // 发送打开信道服务请求，在服务端打开信道
            if(!ret)
            {
                DBG_LOG("openChannel:: open channel failed, channel id %s\n", channel_ptr->getChannelId().c_str());
                return nullptr;
            }
            // DBG_LOG("openChannel:: open channel success, channel id %s\n", channel_ptr->getChannelId().c_str());
            return channel_ptr;
        }

        // 关闭信道
        bool closeChannel(const Channel::ChannelPtr& channel_ptr)
        {
            bool ret = channel_ptr->closeChannel();
            _channel_manager_ptr->removeChannel(channel_ptr->getChannelId());
            return ret;
        }

    private:
        // 接收到通用响应时的回调
        void onCommonResponse(const muduo::net::TcpConnectionPtr &conn, const CommonResponsePtr &rsp, muduo::Timestamp)
        {
            // 找到对应的信道，然后将响应放入到对应的信道异步缓冲区
            Channel::ChannelPtr channel_ptr = _channel_manager_ptr->getChannel(rsp->channel_id());
            if(channel_ptr == nullptr)
            {
                DBG_LOG("not found channel, channel id %s, response id %s\n", rsp->channel_id().c_str(), rsp->response_id().c_str());
                return;
            }
            // DBG_LOG("get common response and put responses, channel id %s, response id %s\n", rsp->channel_id().c_str(), rsp->response_id().c_str());
            channel_ptr->putCommonResponse(rsp); // 将响应放入到responses中，等待请求时的阻塞获取
        }

        // 接收到消息推送时的回调
        void onConsumeResponse(const muduo::net::TcpConnectionPtr &conn, const ConsumeResponsePtr &rsp, muduo::Timestamp)
        {
            // 找到对应的信道，然后将处理消息的函数包装成任务放入线程池
            Channel::ChannelPtr channel_ptr = _channel_manager_ptr->getChannel(rsp->channel_id());
            if(channel_ptr == nullptr)
            {
                DBG_LOG("onConsumeResponse:: not found channel, channel id %s, consumer tag %s\n", rsp->channel_id().c_str(), rsp->consumer_tag().c_str());
                return;
            }

            // DBG_LOG("get message push, channel id: %s, consumer id:%s\n", rsp->channel_id().c_str(), rsp->consumer_tag().c_str());
            _async_worker_ptr->_threadpool.push([channel_ptr, rsp](){
                channel_ptr->consumeTask(rsp);
            });
        }

        // 当连接建立和断开时的回调
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                _conn = conn;
                _downlatch.countDown();
            }
            else
            {
                _conn.reset();
            }
        }

        // 对于未知响应的回调函数
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            conn->shutdown();
        }

    private:
        AsyncWorker::AsyncWorkerPtr _async_worker_ptr;          // 异步工作者线程句柄
        muduo::net::TcpClient _client;                          // muduo库客户端
        muduo::net::TcpConnectionPtr _conn;                     // 连接句柄
        muduo::CountDownLatch _downlatch;                       // 用于控制连接的同步
        ProtobufDispatcher _dispatcher;                         // 请求分发器
        ProtobufCodecPtr _codec_ptr;                            // 协议处理器句柄
        ChannelManager::ChannelManagerPtr _channel_manager_ptr; // 信道管理器句柄
    };
}